package concurrent.completationstage.realexample;

import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;

import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import org.testng.annotations.Test;

public class CSBase {

    static Random random = new Random();

    public static String delayedUpperCase(String s) {
        randomSleep();
        return s.toUpperCase();
    }

    /**
     * 转换为小写
     * @param s
     * @return
     */
    private static String delayedLowerCase(String s) {
        randomSleep();
        return s.toLowerCase();
    }

    static void sleepEnough(){
        try{
            Thread.sleep(2000);
        }catch (InterruptedException e){
            e.printStackTrace();
        }
    }

    static void randomSleep(){
        try {
            Thread.sleep(random.nextInt(1000));
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    static boolean isUpperCase(String s) {
        for (int i = 0; i < s.length(); i++) {
            if (Character.isLowerCase(s.charAt(i))) {
                return false;
            }
        }
        return true;
    }


    public static void main(String[] args) {

        //#1，返回结果为字符串message
        completedFutureExample();

        //#2，可以不运行
//        runAsyncExample();

        //#3，将#1的结果转换成大写字母
//        thenApplyExample();

        //#4,通过调用异步方法，串联起来的completableFuture可以异步执行
//        thenApplyAsyncExample();

        //#5
//        thenApplyAsyncWithExecutorExample();

        //#6，
        thenAcceptExample();

        //#7
        thenAcceptAsyncExample();

//        try {
//            allOfAsyncExample();
//        } finally {
//            executor.shutdown();
//        }
    }

    /**
     * 1、创建一个完成的CompletionFuture
     * 创建一个完成的future有什么用？
     */
    static void completedFutureExample() {

        CompletableFuture<String> cf = CompletableFuture.completedFuture("message");

        assertTrue(cf.isDone());
        //getNow方法在future完成的情况下会返回结果，否则返回传入的参数，这里是null
        assertEquals("message", cf.getNow(null));
    }

    /**
     * 2、运行一个简单的异步阶段
     * 在没有指定executor的情况下，异步执行通过ForkJoinPool实现
     */
    static void runAsyncExample() {
        CompletableFuture<Void> cf = CompletableFuture.runAsync(() -> {
            assertTrue(Thread.currentThread().isDaemon());
            randomSleep();
        });
        assertFalse(cf.isDone());
        sleepEnough();
        assertTrue(cf.isDone());
    }

    /**
     * 3、在上一个阶段上应用函数
     *
     */
    static void thenApplyExample() {
        CompletableFuture<String> cf = CompletableFuture.completedFuture("message")
            .thenApply(s -> {
                assertFalse(Thread.currentThread().isDaemon());
                return s.toUpperCase();
            });
        // 函数的执行会被阻塞，这意味着getNow只有大写操作被完成后才返回
        assertEquals("MESSAGE", cf.getNow(null));
    }

    /**
     * 4、在上一个阶段上异步应用函数
     */
    static void thenApplyAsyncExample() {
        CompletableFuture<String> cf = CompletableFuture.completedFuture("message")
            .thenApplyAsync(s -> {
                assertTrue(Thread.currentThread().isDaemon());
                randomSleep();
                return s.toUpperCase();
        });
        assertNull(cf.getNow(null));
        assertEquals("MESSAGE", cf.join());
    }

    /**
     * 5、使用定制的Executor在前一个阶段上异步应用函数
     * 异步方法一个非常有用的特性就是能够提供一个Executor来异步执行CompletableFuture
     */
    static ExecutorService executor = Executors.newFixedThreadPool(3, new ThreadFactory() {
        int count = 1;

        @Override
        public Thread newThread(Runnable runnable) {
            return new Thread(runnable, "custom-executor-" + count++);
        }
    });

    static void thenApplyAsyncWithExecutorExample() {
        CompletableFuture<String> cf = CompletableFuture.completedFuture("message")
            .thenApplyAsync(s -> {
                assertTrue(Thread.currentThread().getName().startsWith("custom-executor-"));
                assertFalse(Thread.currentThread().isDaemon());
                randomSleep();
                return s.toUpperCase();
            }, executor);

        assertNull(cf.getNow(null));
        assertEquals("MESSAGE", cf.join());
    }

    /**
     * 6：消费前一阶段的结果 </br>
     * 如果下一阶段接收了当前阶段的结果，但是计算的时候不返回值，那么可以不应用一个函数，而是一个消费者
     */
    static void thenAcceptExample() {
        StringBuilder result = new StringBuilder();
        CompletableFuture.completedFuture("thenAccept message")
                .thenAccept(result::append);
        assertTrue(result.length() > 0,"Result was empty" );
    }

    /**
     * 7、异步地消费迁移阶段的结果
     */
    static void thenAcceptAsyncExample() {
        StringBuilder result = new StringBuilder();
        CompletableFuture<Void> cf = CompletableFuture.completedFuture("thenAcceptAsync message")
                .thenAcceptAsync(result::append);

        cf.join();
        assertTrue(result.length() > 0,"Result was empty" );
    }

    /**
     * 8、完成计算异常
     */
    static void completeExceptionallyExample() {
        CompletableFuture<String> cf = CompletableFuture.completedFuture("message")
            .thenApplyAsync(String::toUpperCase,
                CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS)
            );

        CompletableFuture<String> exceptionHandler = cf.handle((s, th) -> {
            return (th != null) ? "message upon cancel" : ""; });

        cf.completeExceptionally(new RuntimeException("completed exceptionally"));
        assertTrue(cf.isCompletedExceptionally(),"Was not completed exceptionally");
        try {
            cf.join();
            fail("Should have thrown an exception");
        } catch(CompletionException ex) { // just for testing
            assertEquals("completed exceptionally", ex.getCause().getMessage());
        }

        assertEquals("message upon cancel", exceptionHandler.join());
    }

    /**
     * 9、取消计算
     */
    static void cancelExample() {
        CompletableFuture<String> cf = CompletableFuture.completedFuture("message")
            .thenApplyAsync(String::toUpperCase,
                CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS));

        CompletableFuture<String> cf2 = cf.exceptionally(throwable -> "canceled message");

        assertTrue( cf.cancel(true),"Was not canceled");
        assertTrue( cf.isCompletedExceptionally(),"Was not completed exceptionally");
        assertEquals("canceled message", cf2.join());
    }

    /**
     * 10、在两个完成的阶段其中之一上应用函数
     */
    static void applyToEitherExample() {
        String original = "Message";
        CompletableFuture<String> stage1 = CompletableFuture.completedFuture(original)
                .thenApplyAsync(CSBase::delayedUpperCase);

        CompletableFuture<String> stage2 = stage1.applyToEither(
            CompletableFuture.completedFuture(original)
                .thenApplyAsync(CSBase::delayedLowerCase),
            s -> s + " from applyToEither"
        );

        assertTrue(stage2.join().endsWith(" from applyToEither"));
    }

    /**
     * 11、在两个完成的阶段其中之一上调用消费函数
     */
    static void acceptEitherExample() {
        String original = "Message";
        StringBuilder result = new StringBuilder();
        CompletableFuture<Void> cf = CompletableFuture.completedFuture(original)
                .thenApplyAsync(CSBase::delayedUpperCase)
                .acceptEither(CompletableFuture.completedFuture(original).thenApplyAsync(CSBase::delayedLowerCase),
                        s -> result.append(s).append("acceptEither")
                );

        cf.join();
        assertTrue( result.toString().endsWith("acceptEither"),"Result was empty");
    }

    /**
     * 12、在两个阶段都执行完成后运行一个Runnable
     */
    static void runAfterBothExample() {
        String original = "Message";
        StringBuilder result = new StringBuilder();
        CompletableFuture.completedFuture(original)
            .thenApply(String::toUpperCase)

            .runAfterBoth(
                CompletableFuture.completedFuture(original).thenApply(String::toLowerCase),
                () -> result.append("done")
            );
        assertTrue(result.length() > 0,"Result was empty");
    }

    /**
     * 13、使用两个BiConsumer消费两个阶段的结果
     */
    static void thenAcceptBothExample() {
        String original = "Message";
        StringBuilder result = new StringBuilder();
        CompletableFuture.completedFuture(original)
            .thenApply(String::toUpperCase)
            .thenAcceptBoth(
                CompletableFuture.completedFuture(original).thenApply(String::toLowerCase),
                (s1, s2) -> result.append(s1 + s2)
            );
        assertEquals("MESSAGEmessage", result.toString());
    }

    /**
     * 14、使用BiFunction 处理两个阶段的结果
     */
    static void thenCombineExample() {
        String original = "Message";
        CompletableFuture<String> cf = CompletableFuture.completedFuture(original).thenApply(s -> delayedUpperCase(s))
                .thenCombine(CompletableFuture.completedFuture(original).thenApply(s -> delayedLowerCase(s)),
                        (s1, s2) -> s1 + s2);
        assertEquals("MESSAGEmessage", cf.getNow(null));
    }

    /**
     * 15、异步使用BiFunction处理两个阶段的结果
     */
    static void thenCombineAsyncExample() {
        String original = "Message";
        CompletableFuture<String> cf = CompletableFuture.completedFuture(original)
                .thenApplyAsync(s -> delayedUpperCase(s))
                .thenCombine(CompletableFuture.completedFuture(original).thenApplyAsync(s -> delayedLowerCase(s)),
                        (s1, s2) -> s1 + s2);
        assertEquals("MESSAGEmessage", cf.join());
    }

    /**
     * 16、组合CompletableFuture
     */
    static void thenComposeExample() {
        String original = "Message";
        CompletableFuture<String> cf = CompletableFuture.completedFuture(original)
            .thenApply(s -> delayedUpperCase(s))

            .thenCompose(upper -> CompletableFuture.completedFuture(original).thenApply(s -> delayedLowerCase(s))
                        .thenApply(s -> upper + s));

        String res = cf.join();
        System.out.println(res);

        //      get方法之外，join方法也会返回结果
        assertEquals("MESSAGEmessage", res);
    }

    /**
     *
     */
    @Test
    private void t16() {
        thenComposeExample();

    }

    /**
     * 17、当几个阶段中的一个完成，创建一个完成的阶段
     */
    static void anyOfExample() {
        StringBuilder result = new StringBuilder();
        List<String> messages = Arrays.asList("a", "b", "c");

        List<CompletableFuture<String>> futures = messages.stream()
                .map(msg -> CompletableFuture.completedFuture(msg).thenApply(CSBase::delayedUpperCase))
                .collect(Collectors.toList());

//      输出 A B C
        futures.forEach(f -> System.out.println(f.getNow("undone")));

//      输出内容 java.util.concurrent.CompletableFuture@6743e411[Completed normally].......
        System.out.println(futures);

        CompletableFuture.anyOf(futures.toArray(new CompletableFuture[futures.size()]))
//           res：上次任务的结果，throwable：上次任务的异常
            .whenComplete((res, throwable) -> {
//              输出  A, throwable :null，A是其中一个future的计算结果
                System.out.println(res + ", throwable :" + throwable);

                if(throwable == null) {
                    assertTrue(isUpperCase((String) res));
                    result.append(res);
                }
            });
        assertTrue(result.length() > 0,"Result was empty");
    }

    @Test
    private void t17() {
        anyOfExample();
    }

    /**
     * 18、当所有的阶段都完成后同步创建一个阶段
     */
    static void allOfExample() {
        StringBuilder result = new StringBuilder();
        List<String> messages = Arrays.asList("a", "b", "c");

//      会有a、b、c 3个future; future可以存储在集合中
        List<CompletableFuture<String>> futures = messages.stream()
                .map(msg -> CompletableFuture.completedFuture(msg).thenApply(CSBase::delayedUpperCase))
                .collect(Collectors.toList());

        futures.forEach(f -> System.out.println(f.getNow("undone")));

        System.out.println(futures.size());

//      allOf没有返回值，这是因为每个传入的 CompletableFuture 的返回值都可能不同，所以【组合的结果】是无法用某种类型来表示的，索性返回 Void 类型
        CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()]))
            .whenComplete((res, throwable) -> {
                // null, null，因为allOf接收的future的返回值是不确定的，无法用一个确定的
                System.out.println("res " + res + ", throwable :" + throwable);
                futures.forEach(cf -> assertTrue(isUpperCase(cf.getNow(null))));
                result.append("done");

            });

        assertTrue( result.length() > 0,"Result was empty");
    }

    @Test
    private void t18() {
        allOfExample();
    }

    /**
     * 19、当所有的阶段都完成后异步地创建一个阶段
     */
    static void allOfAsyncExample() {
        StringBuilder result = new StringBuilder();
        List<String> messages = Arrays.asList("a", "b", "c");
        List<CompletableFuture<String>> futures = messages.stream()
                .map(msg -> CompletableFuture.completedFuture(msg).thenApplyAsync(s -> delayedUpperCase(s)))
                .collect(Collectors.toList());

        CompletableFuture<Void> allOf = CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()]))
                .whenComplete((v, throwable) -> {
                    futures.forEach(cf -> assertTrue(isUpperCase(cf.getNow(null))));
                    result.append("done");
                });
        allOf.join();
        assertTrue(result.length() > 0,"Result was empty");
    }




}
